package com.dabai.iot.mqtt.event;

import com.dabai.common.core.utils.Func;
import com.dabai.iot.core.api.RemoteConsumerService;
import com.dabai.iot.core.entity.ConsumerGroupEntity;
import com.dabai.iot.mqtt.model.ServerClient;
import com.dabai.iot.mqtt.model.ServerConnectionSession;
import com.dabai.iot.mqtt.util.ClientUtil;
import com.dabai.iot.mqtt.util.ServerConnectionSessionManager;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.vertx.mqtt.MqttEndpoint;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 服务端（业务平台）连接状态监听
 *
 * @author kev1n
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ServerConnectStatusListener implements IMqttConnectStatusListener {

    @DubboReference(stub = "true")
    private final RemoteConsumerService remoteConsumerService;

    /**
     * 服务端上线
     */
    @Override
    public boolean online(MqttEndpoint endpoint) {
        final String clientId = endpoint.clientIdentifier();
        ServerClient serverClient = ClientUtil.parseClientId(clientId);
        String tenantId = serverClient.getTenantId();
        String groupID = serverClient.getGroupID();

        // 验证消费组
        ConsumerGroupEntity consumerGroup = remoteConsumerService.getConsumerByGroupID(groupID);
        if (consumerGroup == null) {
            log.info("MQTT Server ====> 服务端订阅 [{}] 订阅消费组不存在", clientId);
            endpoint.reject(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED);
            return false;
        }

        // accept connection from the remote client
        return acceptServerSubscription(endpoint, tenantId, groupID);
    }

    /**
     * 服务端离线
     */
    @Override
    public void offline(MqttEndpoint endpoint) {
        final String clientId = endpoint.clientIdentifier();

        ServerClient serverClient = ClientUtil.parseClientId(clientId);
        String groupID = serverClient.getGroupID();

        List<ServerConnectionSession> sessionList = ServerConnectionSessionManager.getInstance().getSession(groupID);

        if (Func.isNotEmpty(sessionList)) {
            for (ServerConnectionSession session : sessionList) {
                if (session.getMqttEndpoint().clientIdentifier().equals(endpoint.clientIdentifier())) {
                    // 服务端订阅断开只需清除会话，无需其他操作
                    ServerConnectionSessionManager.getInstance().removeSession(groupID, session);
                }
            }
        }
    }

    /**
     * 接受有效服务端订阅
     * <p>
     * 1 服务端连接时自动订阅主题 $SYS/tenantId/${groupID}/status 接收上下线状态
     * 2 服务端连接时自动订阅主题 $SYS/tenantId/${groupID}/update 接收数据上报
     */
    private boolean acceptServerSubscription(MqttEndpoint endpoint, String tenantId, String groupID) {
        endpoint.accept();

        // 保存连接会话
        ServerConnectionSession session = new ServerConnectionSession(tenantId, endpoint);
        ServerConnectionSessionManager.getInstance().addSession(groupID, session);

        return true;
    }

}
